package team.bluepen.supermarket.service.calc

import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.reflect.io.Path


/**
 *
 * @author YCB 2022/12/17 20:15
 *
 */
class ProductSparkCoreStatus extends Serializable {

  def productStatusStatistics(path: Path): mutable.Iterable[SupermarketProduct] = {

    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ProductStatus")
    val sc = new SparkContext(sparkConf)

    val actionRDD = sc.textFile("data/supermarket_product") // true:上架 false:下架

    val acc = new SupermarketProductAccumulator
    sc.register(acc, "supermarketProduct")

    // 数据结构转换
    actionRDD.foreach(
      action => {
        val data = action.split(" ")
        if (data(4) == "true") { // 上架
          acc.add(((data(0), data(1)), 1, 0))
        } else if (data(4) == "false") { // 下架
          acc.add(((data(0), data(1)), 0, 1))
        }
      }
    )

    sc.stop()
    val accVal = acc.value

    //    ((11,book),SupermarketProduct(111,Book,1,2)) => (111,book,1,2)
    accVal.map(_._2)
  }

  // 样例类在编译时，会自动混入序列化特质（实现可序列化接口）
  case class SupermarketProduct(productId: String, productName: String, var shelves: Int, var shelvesOff: Int)

  // ((111,Book),1,2)
  class SupermarketProductAccumulator extends AccumulatorV2[((String, String), Int, Int), mutable.Map[(String, String), SupermarketProduct]] {
    private val spMap = mutable.Map[(String, String), SupermarketProduct]()

    override def isZero: Boolean = {
      spMap.isEmpty
    }

    override def copy(): AccumulatorV2[((String, String), Int, Int), mutable.Map[(String, String), SupermarketProduct]] = {
      new SupermarketProductAccumulator
    }

    override def reset(): Unit = {
      spMap.clear()
    }

    override def add(v: ((String, String), Int, Int)): Unit = {
      val productInfo: (String, String) = v._1
      val supermarketProduct: SupermarketProduct = spMap.getOrElse(productInfo, SupermarketProduct(productInfo._1, productInfo._2, 0, 0))
      supermarketProduct.shelves += v._2
      supermarketProduct.shelvesOff += v._3

      spMap.update(productInfo, supermarketProduct)
    }

    override def merge(other: AccumulatorV2[((String, String), Int, Int), mutable.Map[(String, String), SupermarketProduct]]): Unit = {
      val mapThis: mutable.Map[(String, String), SupermarketProduct] = this.spMap
      val mapOther: mutable.Map[(String, String), SupermarketProduct] = other.value

      mapOther.foreach {
        case (productInfo, supermarketProduct) => {
          var productNew: SupermarketProduct = mapThis.getOrElse(productInfo, SupermarketProduct(productInfo._1, productInfo._2, 0, 0))
          productNew.shelves += supermarketProduct.shelves
          productNew.shelvesOff += supermarketProduct.shelvesOff
          mapThis.update(productInfo, productNew)
        }
      }
    }

    override def value: mutable.Map[(String, String), SupermarketProduct] = {
      spMap
    }
  }
}
